package com.bw.gmall.realtime.utils;

import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.kafka.connect.json.JsonConverterConfig;

import java.util.HashMap;
import java.util.Map;

public class MysqlUtil {

    public static String getBaseDicLookUpDDL() {
        return "create table `base_dic`( " +
                "`dic_code` string, " +
                "`dic_name` string, " +
                "`parent_code` string, " +
                "`create_time` timestamp, " +
                "`operate_time` timestamp, " +
                "primary key(`dic_code`) not enforced " +
                ")" + MysqlUtil.mysqlLookUpTableDDL("base_dic");
    }

    public static String mysqlLookUpTableDDL(String tableName) {

        return " WITH ( " +
                "'connector' = 'jdbc', " +
                "'url' = 'jdbc:mysql://hadoop102:3306/gmall', " +
                "'table-name' = '" + tableName + "', " +
                "'lookup.cache.max-rows' = '10', " +
                "'lookup.cache.ttl' = '1 hour', " +
                "'username' = 'root', " +
                "'password' = 'root', " +
                "'driver' = 'com.mysql.cj.jdbc.Driver' " +
                ")";
    }

    /**
     * 使用Flink CDC方式，拉取Mysql表数据，先全量加载，再增量读取
     * @param env 流式执行环境
     * @param database 数据库名称
     * @param table 表名称
     * @return 数据流，数据类型为json字符串
     */
    public static DataStream<String> cdcMysql(StreamExecutionEnvironment env, String database, String table){
        // a. 反序列化
        Map<String, Object> customConverterConfigs = new HashMap<>();
//		使用JsonDebeziumDeserializationSchema
//		不配置customConverterConfigs.put(JsonConverterConfig.DECIMAL_FORMAT_CONFIG, "numeric")的话，
        customConverterConfigs.put(JsonConverterConfig.DECIMAL_FORMAT_CONFIG, "numeric");
//		decimal 会解析成错误的字符串，
        JsonDebeziumDeserializationSchema schema =
                new JsonDebeziumDeserializationSchema();
        // b. 数据源
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("hadoop102")
                .port(3306)
                .databaseList(database)
                .tableList(database + "." + table)
                .username("root")
                .password("root")
                //从BinLog第一行数据开始读, 最好先给这个数据库加上BinLog后, 再去读取创建数据库；
                //1.历史数据都要initial
                //2.还是只读最新的
                .startupOptions(StartupOptions.initial())
                .deserializer(schema)
                .build();
        // c. 读取数据
        DataStreamSource<String> stream = env.fromSource(
                mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source"
        );
        // d. 返回
        return stream;
    }


    /**
     * 将DataStream数据流中数据保存到MySQL表，要求数据类型为String，各个字段之间值使用逗号隔开
     * @param stream 数据流，数据类型为String
     * @param databaseName 数据库名称
     * @param tableName 表名称
     * @param columnNames 列名称，各个列名称之间逗号隔开
     */
    public static void sinkToMysql(DataStream<String> stream, String databaseName,
                                   String tableName, String columnNames){
        // a. 插入语句
        int length = columnNames.split(",").length;
        String values = "" ;
        for(int i = 0 ; i < length; i ++){
            values = values + "?, " ;
        }
        String insertSql = "INSERT INTO " + databaseName + "." + tableName
                + "(" + columnNames +") VALUES ("
                + values.substring(0, values.length() - 2)
                +")";
        // b. sink对象
        SinkFunction<String> sink = JdbcSink.<String>sink(
                insertSql,
                (statement, value) -> {
                    String[] split = value.split(",");
                    for(int i = 0; i < split.length; i ++){
                        statement.setObject(i + 1, split[i]);
                    }
                },
                JdbcExecutionOptions.builder()
                        .withBatchSize(1000)
                        .withBatchIntervalMs(200)
                        .withMaxRetries(5)
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withUrl("jdbc:mysql://node101:3306/?useSSL=false&useUnicode=true&characterEncoding=UTF-8")
                        .withDriverName("com.mysql.cj.jdbc.Driver")
                        .withUsername("root")
                        .withPassword("123456")
                        .build()
        );
        // c. 保存数据
        stream.addSink(sink);
    }
}